package com.hliushi.process

import java.sql.DriverManager

import com.alibaba.fastjson.JSON
import com.hliushi.utils.OffsetUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * descriptions: 疫情物资数据的实时处理与分析
 *
 * author: Hliushi
 * date: 2021/7/1 15:56
 */
object Covid19_WZData_Process {

  /**
   * Spark Streaming, 提供了基于RDDs的DStream API, 每个时间间隔内的数据为一个RDD, 源源不断对RDD进行处理来实现流计算
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {
    // 1.准备SparkStreaming的开发环境
    val conf = new SparkConf().setAppName("covid19_wzdata_process").setMaster("local[6]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val ssc = new StreamingContext(sc, Seconds(5))
    ssc.checkpoint("./sscckp")

    // 补充知识点: SparkStreaming整合Kafka的两种方式:
    /**
     * 1.Receiver模式
     * KafkaUtils.createDStream()  -- API创建
     * 会有一个Receiver作为常驻Task运行在Executor进程中, 一直等待数据的到来 [推模式 ?]
     * 一个Receiver效率会比较低, 那么可以使用多个Receiver, 但是多个Receiver中的数据又需要手动进行Union(合并)很麻烦
     * 且其中某个Receiver挂了, 会导致数据丢失, 需要开启WAL预写日志来保证数据安全, 但是效率又低了
     * Receiver模式使用Zookeeper连连接Kafka (Kafka的新版本已经不推荐使用Zookeeper了)
     * Receiver模式使用的是Kafka的高阶API(高度封装的), offset由Receiver提交到ZK中
     * (Kafka的新版本中offset默认存储在默认主题_consumer_offset中的, 不推荐存入到ZK中了)
     * 容易和Spark维护在Checkpoint中的offset不一致
     * 所以不管从何种角度去说Receiver模式已经不在合适如今的Kafka版本了, 面试需要说出以上的原因!
     */

    /**
     * 2.Direct模式
     * KafkaUtils.createDirectStream() -- API创建
     * Direct模式是直接连接Kafka的各个分区, 并拉取数据, 提高了数据读取的并发能力
     * Direct模式使用的是Kafka低阶API(底层API), 可以自己维护偏移量到任何地方
     * (默认是由Spark提交到默认主题/Checkpoint)
     * Direct模式 + 手动操作可以保证数据的Exactly-Once精准一次(数据仅会被处理一次)
     */
    // 补充知识点: SparkStreaming整合Kafka的两个版本的API
    // spark-streaming-kafka-0-8
    // 支持Receiver模式和Direct模式, 但是不支持offset维护API, 不支持动态分区订阅

    // spark-streaming-kafka-0-10
    // 支持Direct, 不支持Receiver模式, 支持offset维护API, 支持动态分区订阅
    // 结论: 使用spark-streaming-kafka-0-10即可


    // 2.准备kafka的连接参数
    //val kafkaParams: Map[String, Object] = Map[String, Object](
    //  "bootstrap.servers" -> GlobalConfigUtil.bootstrapServers,
    //  "group.id" -> GlobalConfigUtil.groupId,
    //  "enable.auto.commit" -> GlobalConfigUtil.enableAutoCommit,
    //  "auto.commit.interval.ms" -> GlobalConfigUtil.autoCommitIntervalMs,
    //  "auto.offset.reset" -> GlobalConfigUtil.autoOffsetReset,
    //  //"key.deserializer" -> GlobalConfigUtil.keyDeserializer.getClass,
    //  "key.deserializer" -> classOf[StringDeserializer],
    //  //"value.deserializer" -> Class.forName(GlobalConfigUtil.valueDeserializer).getClass
    //  "value.deserializer" -> classOf[StringDeserializer]
    //)
    val kafkaParams: Map[String, Object] = Map[String, Object](
      "bootstrap.servers" -> "cvm-node01:9092,cvm-node02:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "SparkKafka",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean) // 是否自动维护偏移量
    )

    val topics = Array("covid19_wz")

    // 从MySQL中查询出offset:Map[TopicPartition, Long]
    val offsetsMap: Map[TopicPartition, Long] = OffsetUtils.getOffsetMap("SparkKafka", "covid19_wz")


    val kafkaDS = if (offsetsMap.nonEmpty) {
      println("Mysql记录了offset信息, 从offset出开始消费")

      // 3.连接kafka获取消息
      KafkaUtils.createDirectStream[String, String](
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, offsetsMap))
    } else {
      println("Mysql没有记录offset信息, 从latest出开始消费")

      KafkaUtils.createDirectStream[String, String](
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))
    }


    // 4.实时处理数据并手动提交偏移量
    // val valueDS = kafkaDS.map(x => x.value())
    // valueDS.print()
    // 我们要手动提交偏移量, 那么就意味着, 消费了一批数据就应该提交一次偏移量
    // 在SparkStreaming中数据抽象为DStream, DStream的底层其实也是RDD, 也就是每一批次的数据
    // 所以接下来我们应该对DStream中的RDD进行处理
    kafkaDS.foreachRDD((rdd: RDD[ConsumerRecord[String, String]]) => {
      // 如果该rdd中有数据则处理
      if (rdd.count() > 0) {
        // rdd.foreach((record: ConsumerRecord[String, String]) => println("从Kafka中消费到的每一条消息:" + record))
        // ConsumerRecord(topic = covid19_wz, partition = 1, offset = 76, CreateTime = 1625131474822, checksum = 824588334, serialized key size = -1, serialized value size = 3, key = null, value = 234)
        // 获取偏移量
        // 使用Spark-streaming-kafka-0-10中封装好的API来存放偏移量并提交
        val offsets = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        //for (offset <- offsets) {
        //  println(s"topic = ${offset.topic}, partition = ${offset.partition}, fromOffset = ${offset.fromOffset}, " +
        //    s"until = ${offset.untilOffset}")
        //  // topic = covid19_wz, partition = 1, fromOffset = 76, until = 89
        //  // topic = covid19_wz, partition = 0, fromOffset = 122, until = 135
        //  // -------------------------------------------------------------
        //  // topic = covid19_wz, partition = 1, fromOffset = 89, until = 89
        //  // topic = covid19_wz, partition = 0, fromOffset = 135, until = 136
        //  // -------------------------------------------------------------
        //  // topic = covid19_wz, partition = 1, fromOffset = 89, until = 90
        //  // topic = covid19_wz, partition = 0, fromOffset = 136, until = 136
        //}

        // 手动提交偏移量到Kafka的默认主题: __consumer__offsets中, 如果开启了Checkpoint还会提交到Checkpoint中
        //kafkaDS.asInstanceOf[CanCommitOffsets].commitAsync(offsets)
        OffsetUtils.saveOffsets("SparkKafka", offsets)
      }
    })

    // 需求分析:
    // {"count":843,"from":"消耗","name":"N95口罩/个"}
    // 我们从Kafka中消费的数据为如上格式的jsonStr, 需要解析为JSON对象(或者是样例类)
    // 目标是:将数据转为如下格式
    // 名称, 采购, 下拨, 捐赠, 消耗, 需求, 库存
    // N95口罩/个, 1000, 1000, 500, -1000, -1000, 500
    // 为了达成目标结果, 我们需要对每一条数据进行处理, 得出如下格式:
    // (name, (采购, 下拨, 捐赠, 消耗, 需求, 库存))
    // 如:接收到一条数据为:
    // {"count":843,"from":"消耗","name":"N95口罩/个"}
    // 应该记为:
    // (N95口罩/个, (采购0, 下拨0, 捐赠0, 消耗-843, 需求0, 库存-843))
    // 再来一条数据:
    // {"count":500,"from":"采购","name":"N95口罩/个"}
    // 应该记为:
    // (N95口罩/个, (采购500, 下拨0, 捐赠0, 消耗0, 需求0, 库存500))
    // 最后聚合结果:
    // (N95口罩/个, (采购500, 下拨0, 捐赠0, 消耗-843, 需求0, 库存-343))

    // 4.1.将接收到的数据转换为需要的元组格式 (name, (采购, 下拨, 捐赠, 消耗, 需求, 库存))
    /**
     * bug点, 这里没有显式写出返回参数
     * 导致默认给出的返回值为 DStream[(String, (Any, Any, Any, Int, Int, Any))]
     * 显示给出 返回值类型DStream[(String, (Int, Int, Int, Int, Int, Int))]就行了
     */
    val tupleDS: DStream[(String, (Int, Int, Int, Int, Int, Int))] = kafkaDS.map((record: ConsumerRecord[String, String]) => {
      val jsonStr = record.value()
      val jSONObj = JSON.parseObject(jsonStr)
      val name = jSONObj.getString("name")
      val from = jSONObj.getString("from")
      val count = jSONObj.getInteger("count")

      // 根据物资来源不同, 将count记在不同的位置, 最终形成统一的格式
      from match {
        case "采购" => (name, (count, 0, 0, 0, 0, count))
        case "下拨" => (name, (0, count, 0, 0, 0, count))
        case "捐赠" => (name, (0, 0, count, 0, 0, count))
        case "消耗" => (name, (0, 0, 0, -count, 0, -count))
        case "需求" => (name, (0, 0, 0, 0, -count, -count))
      }
    })

    //tupleDS.print()
    // (一次性手套/副,(0,0,0,-29,0,-29))
    // (护目镜/副,(739,0,0,0,0,739))
    // (医用防护服/套,(0,0,0,0,-584,-584))
    // (84消毒液/瓶,(627,0,0,0,0,627))

    // 4.2将上述格式的数据按照key进行聚合(有状态的计算) --使用updateStateByKey
    // 定义一个函数, 用来将当前批次的数据和历史数据进行聚合
    val updateFunc = (currentValues: Seq[(Int, Int, Int, Int, Int, Int)], historyValue: Option[(Int, Int, Int, Int, Int, Int)]) => {
      // 0.定义变量用来接收当前批次数据(采购, 下拨, 捐赠, 消耗, 需求, 库存)
      var current_cg: Int = 0
      var current_xb: Int = 0
      var current_jz: Int = 0
      var current_xh: Int = 0
      var current_xq: Int = 0
      var current_kc: Int = 0

      if (currentValues.nonEmpty) {
        // 1.取出当前批次数据
        for (value <- currentValues) {
          current_cg += value._1
          current_xb += value._2
          current_jz += value._3
          current_xh += value._4
          current_xq += value._5
          current_kc += value._6
        }

        // 2.取出历史数据
        val history_cg = historyValue.getOrElse((0, 0, 0, 0, 0, 0))._1
        val history_xb = historyValue.getOrElse((0, 0, 0, 0, 0, 0))._2
        val history_jz = historyValue.getOrElse((0, 0, 0, 0, 0, 0))._3
        val history_xh = historyValue.getOrElse((0, 0, 0, 0, 0, 0))._4
        val history_xq = historyValue.getOrElse((0, 0, 0, 0, 0, 0))._5
        val history_kc = historyValue.getOrElse((0, 0, 0, 0, 0, 0))._6

        // 3.将当前批次数据和历史数据进行聚合
        val result_cg: Int = current_cg + history_cg
        val result_xb: Int = current_xb + history_xb
        val result_jz: Int = current_jz + history_jz
        val result_xh: Int = current_xh + history_xh
        val result_xq: Int = current_xq + history_xq
        val result_kc: Int = current_kc + history_kc

        Some((
          result_cg,
          result_xb,
          result_jz,
          result_xh,
          result_xq,
          result_kc))
      } else {
        historyValue // 如果当前批次没有数据直接返回之前的值即可
      }
    }

    val resultDS: DStream[(String, (Int, Int, Int, Int, Int, Int))] = tupleDS.updateStateByKey(updateFunc)
    resultDS.print()

    // 5.将处理分析的结果存入到MySQL
    //  CREATE TABLE `covid19_wz` (
    //  `name` varchar(12) NOT NULL DEFAULT '',
    //  `cg` int(11) DEFAULT '0',
    //  `xb` int(11) DEFAULT '0',
    //  `jz` int(11) DEFAULT '0',
    //  `xh` int(11) DEFAULT '0',
    //  `xq` int(11) DEFAULT '0',
    //  `kc` int(11) DEFAULT '0',
    //  PRIMARY KEY (`name`)
    //) ENGINE=InnoDB DEFAULT CHARSET=utf8;
    resultDS.foreachRDD((rdd: RDD[(String, (Int, Int, Int, Int, Int, Int))]) => {
      rdd.foreachPartition(lines => {
        // 1.开启连接
        val conn = DriverManager.getConnection("jdbc:mysql://cvm-node01:3306/bigdata?characterEncoding=UTF-8&useSSL=false", "root", "123456")
        // 2.编写SQL并获取ps
        val sql =
          """
            |replace into covid19_wz(name, cg, xb, jz, xh, xq, kc)
            |values(?, ?, ?, ?, ?, ?, ?)
            |""".stripMargin

        val ps = conn.prepareStatement(sql)

        // 3.设置参数并执行
        for (line <- lines) {
          ps.setString(1, line._1)
          ps.setInt(2, line._2._1)
          ps.setInt(3, line._2._2)
          ps.setInt(4, line._2._3)
          ps.setInt(5, line._2._4)
          ps.setInt(6, line._2._5)
          ps.setInt(7, line._2._6)
          ps.executeUpdate()
        }

        // 4.关闭资源
        ps.close()
        conn.close()
      })
    })

    // 6.开启SparkStreaming任务并等待结束
    ssc.start()
    ssc.awaitTermination()
  }
}